package com.aaron.registry.zk.util;

import com.aaron.enums.RpcConfigEnum;
import com.aaron.utils.PropertiesFileUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Curator(zookeeper client) utils，ZKjava客户端工具类
 * 一个znode节点：
 */
@Slf4j
public final class CuratorUtils {
    // 重试之间等待的初始时间
    private static final int BASE_SLEEP_TIME = 1000;
    // 最大重试次数
    private static final int MAX_RETRIES = 3;
    // zk注册路径
    public static final String ZK_REGISTER_ROOT_PATH = "/my-rpc";
    // 服务地址集合，key:服务名，value:可选服务地址集合
    private static final Map<String, List<String>> SERVICE_ADDRESS_MAP = new ConcurrentHashMap<>();
    // 服务名的集合，newKeySet()：新建一个set集合
    private static final Set<String> REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet();
    // CuratorFramework 是Netflix公司开发一款连接zookeeper服务的框架，提供了比较全面的功能，除了基础的节点的操作，节点的监听，还有集群的连接以及重试。
    private static CuratorFramework zkClient;
    // 默认zk地址
    private static final String DEFAULT_ZOOKEEPER_ADDRESS = "127.0.0.1:2181";

    private CuratorUtils() {
    }

    // 创建持久性节点。与临时节点不同，当客户端断开连接时，不会删除持久性节点
    public static void createPersistentNode(CuratorFramework zkClient, String path) {
        try {
            if (REGISTERED_PATH_SET.contains(path) || zkClient.checkExists().forPath(path) != null) {
                log.info("The node already exists. The node is:[{}]", path);
            } else {
                /**
                 * creatingParentsIfNeeded()方法保证父节点不存在自动创建父节点
                 * withMode(CreateMode.PERSISTENT)方法指定创建节点类型，表示创建持久节点
                 * eg: /my-rpc/github.javaguide.HelloService/127.0.0.1:9999
                 *    /my-rpc/com.aaron.HelloServicetest2version2/192.168.5.1:8081
                 */
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                log.info("The node was created successfully. The node is:[{}]", path);
            }
            // 记得向REGISTERED_PATH_SET的Set集合中添加
            REGISTERED_PATH_SET.add(path);
        } catch (Exception e) {
            log.error("create persistent node for path [{}] fail", path);
        }
    }

    // 获取某个服务节点下面的子节点，list集合里面的都是String类型的，例：192.168.5.1:8081
    public static List<String> getChildrenNodes(CuratorFramework zkClient, String rpcServiceName) {
        // 先从缓存中拿
        if (SERVICE_ADDRESS_MAP.containsKey(rpcServiceName)) {
            return SERVICE_ADDRESS_MAP.get(rpcServiceName);
        }
        List<String> result = null;
        // rpcServiceName eg:github.javaguide.HelloServicetest2version1
        String servicePath = ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName;
        try {
            result = zkClient.getChildren().forPath(servicePath);
            SERVICE_ADDRESS_MAP.put(rpcServiceName, result);
            // 注册器监听对指定节点的更改
            registerWatcher(rpcServiceName, zkClient);
        } catch (Exception e) {
            log.error("get children nodes for path [{}] fail", servicePath);
        }
        return result;
    }

    // 清空指定路径的注册信息
    public static void clearRegistry(CuratorFramework zkClient, InetSocketAddress inetSocketAddress) {
        // parallel()其实就是一个并行执行的流,可能提高你的多线程任务的速度.
        // REGISTERED_PATH_SET是个Set<String>集合
        REGISTERED_PATH_SET.stream().parallel().forEach(p -> {
            try {
                // endswith()方法用于判断字符串是否以指定后缀结尾，如果以指定后缀结尾返回True，否则返回False。
                if (p.endsWith(inetSocketAddress.toString())) {
                    zkClient.delete().forPath(p);
                }
            } catch (Exception e) {
                log.error("clear registry for path [{}] fail", p);
            }
        });
        log.info("All registered services on the server are cleared:[{}]", REGISTERED_PATH_SET.toString());
    }

    // zkClient的连接（连接到zk服务端啊）
    public static CuratorFramework getZkClient() {
        // 读取resources资源目录下的rpc.properties文件，获取到的是地址eg:127.0.0.1:2181
        Properties properties = PropertiesFileUtil.readPropertiesFile(RpcConfigEnum.RPC_CONFIG_PATH.getPropertyValue());
        // getPropertyValue()方法与getProperty()方法。获取zookeeper服务的地址：获取到的是地址eg:127.0.0.1:2181
        String zookeeperAddress = properties != null && properties.getProperty(RpcConfigEnum.ZK_ADDRESS.getPropertyValue()) != null ? properties.getProperty(RpcConfigEnum.ZK_ADDRESS.getPropertyValue()) : DEFAULT_ZOOKEEPER_ADDRESS;
        // 如果zkClient已启动，则直接返回
        if (zkClient != null && zkClient.getState() == CuratorFrameworkState.STARTED) {
            return zkClient;
        }
        // 如果zkClient没有启动，则使用重试策略RetryPolicy，重试3次，这将增加重试之间的睡眠时间。
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        zkClient = CuratorFrameworkFactory.builder()
                // the server to connect to (can be a server list)
                .connectString(zookeeperAddress)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        try {
            // wait 30s until connect to the zookeeper
            if (!zkClient.blockUntilConnected(30, TimeUnit.SECONDS)) {
                throw new RuntimeException("Time out waiting to connect to ZK!");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return zkClient;
    }

    // 注册器以侦听对指定节点的更改
    private static void registerWatcher(String rpcServiceName, CuratorFramework zkClient) throws Exception {
        // rpcServiceName eg:github.javaguide.HelloServicetest2version
        String servicePath = ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName;
        // 子节点监听
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
        // lambda表达式。
        // 先执行 List<String> serviceAddresses = curatorFramework.getChildren().forPath(servicePath);
        // 再执行 SERVICE_ADDRESS_MAP.put(rpcServiceName, serviceAddresses);
        PathChildrenCacheListener pathChildrenCacheListener = (curatorFramework, pathChildrenCacheEvent) -> {
            List<String> serviceAddresses = curatorFramework.getChildren().forPath(servicePath);
            SERVICE_ADDRESS_MAP.put(rpcServiceName, serviceAddresses);
        };
        // 配置并启动子节点监听
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
        pathChildrenCache.start();
    }

}
